package com.bff.gaia.mix.examples.xjoin;
/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import com.bff.gaia.api.common.functions.MapFunction;
import com.bff.gaia.api.common.functions.RichMapFunction;
import com.bff.gaia.api.java.tuple.Tuple1;
import com.bff.gaia.api.java.tuple.Tuple2;
import com.bff.gaia.api.java.tuple.Tuple7;
import com.bff.gaia.api.java.tuple.Tuple8;
import com.bff.gaia.configuration.Configuration;
import com.bff.gaia.mix.api.TimeCharacteristic;
import com.bff.gaia.mix.api.datastream.DataMixStream;
import com.bff.gaia.mix.api.environment.MixStreamExecutionEnvironment;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * Description: gaia-tiger
 * Created by tiger on 2019/4/15 16:56.
 * xjoin.example.mapBenchmark
 */
public class mapBenchmark {
	public static void main(String[] args) throws Exception {
		MixStreamExecutionEnvironment env = MixStreamExecutionEnvironment.getExecutionEnvironment();
		env.setMixStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

//		final String jdbcDriver = "com.mysql.cj.jdbc.Driver";
//		final String jdbcUrl = "jdbc:mysql://192.168.100.203:3306/employees?characterEncoding=UTF-8&useSSL=false&serverTimezone=Hongkong&rewriteBatchedStatements=true&allowPublicKeyRetrieval=true";
//		final String jdbcUser = "root";
//		final String jdbcPassword = "123456";

		final String jdbcDriver="com.mysql.cj.jdbc.Driver";
		final String jdbcUrl="jdbc:mysql://127.0.0.1:3306/employees?characterEncoding=UTF-8&useSSL=false&serverTimezone=Hongkong&rewriteBatchedStatements=true&allowPublicKeyRetrieval=true";
		final String jdbcUser="gaia";
		final String jdbcPassword="labmysqlgaia";

//		生成员工编号（10000-11000） 构建数据流
//		List<Integer> ints = new ArrayList<>();
//		Random random = new Random();
//		int num=1000;
//		if(args.length==1){
//			num=Integer.valueOf(args[0]);
//		}
//		for (int i = 0; i < num; i++) {
//			ints.add(random.nextInt(999) + 10001);
//		}
//		DataStream<Integer> empNoStream =env.fromCollection(ints);

		int num = 1000;
		if (args.length == 1) {
			num = Integer.valueOf(args[0]);
		}
		DataMixStream<Integer> empNoStream = env.addSource(new benchmarkDataSource.SimpleSource(num));

		//转换tuple
		DataMixStream<Tuple1<Integer>> empNoTupleStream = empNoStream.map(new MapFunction<Integer, Tuple1<Integer>>() {
			@Override
			public Tuple1<Integer> map(Integer value) throws Exception {
				return new Tuple1<Integer>(value);
			}
		});

		//连接 员工部门表
		DataMixStream<Tuple2<Integer, String>> dept_empTupleStream = empNoTupleStream.map(new RichMapFunction<Tuple1<Integer>, Tuple2<Integer, String>>() {

			private transient Connection connection = null;
			private transient PreparedStatement ps = null;

			@Override
			public void open(Configuration parameters) throws Exception {
				super.open(parameters);
				Class.forName(jdbcDriver);
				connection = DriverManager.getConnection(jdbcUrl,
						jdbcUser,
						jdbcPassword);
				ps = connection.prepareStatement("select * from dept_emp where emp_no= ?");
			}

			@Override
			public Tuple2<Integer, String> map(Tuple1<Integer> integerTuple1) throws Exception {
				ps.setInt(1, integerTuple1.f0);
				ResultSet resultSet = ps.executeQuery();
				if (resultSet.next()) {
					return new Tuple2<>(
							integerTuple1.f0,
							resultSet.getString("dept_no"));
				} else {
					return null;
				}
			}

			@Override
			public void close () throws Exception{
				ps.close();
				connection.close();
			}
		});

		//扩展员工信息
		DataMixStream<Tuple7<Integer, String, String, String, String, String, String>> empExTupleStream = dept_empTupleStream.map(new RichMapFunction<Tuple2<Integer, String>, Tuple7<Integer, String, String, String, String, String, String>>() {

			private transient Connection connection = null;
			private transient PreparedStatement ps = null;

			@Override
			public void open(Configuration parameters) throws Exception {
				super.open(parameters);
				Class.forName(jdbcDriver);
				connection = DriverManager.getConnection(jdbcUrl,
						jdbcUser,
						jdbcPassword);
				ps = connection.prepareStatement("select * from employees where emp_no= ?");
			}

			@Override
			public Tuple7<Integer, String, String, String, String, String, String> map(Tuple2<Integer, String> integerStringTuple2) throws Exception {
				ps.setInt(1, integerStringTuple2.f0);
				ResultSet resultSet = ps.executeQuery();
				if (resultSet.next()) {
					return new Tuple7<>(
							integerStringTuple2.f0,
							integerStringTuple2.f1,
							String.valueOf(resultSet.getDate("birth_date")),
							resultSet.getString("first_name"),
							resultSet.getString("last_name"),
							resultSet.getString("gender"),
							String.valueOf(resultSet.getDate("hire_date")));
				} else {
					return null;
				}
			}


			@Override
			public void close () throws Exception{
				ps.close();
				connection.close();
			}
		});

		//扩展部门信息
		DataMixStream<Tuple8<Integer, String, String, String, String, String, String, String>> empExProTupleStream = empExTupleStream.map(new RichMapFunction<Tuple7<Integer, String, String, String, String, String, String>, Tuple8<Integer, String, String, String, String, String, String, String>>() {

			private transient Connection connection = null;
			private transient PreparedStatement ps = null;

			@Override
			public void open(Configuration parameters) throws Exception {
				super.open(parameters);
				Class.forName(jdbcDriver);
				connection = DriverManager.getConnection(jdbcUrl,
						jdbcUser,
						jdbcPassword);
				ps = connection.prepareStatement("select * from departments where dept_no= ?");
			}

			@Override
			public Tuple8<Integer, String, String, String, String, String, String, String> map(Tuple7<Integer, String, String, String, String, String, String> integerStringStringStringStringStringStringTuple7) throws Exception {
				ps.setString(1, integerStringStringStringStringStringStringTuple7.f1);
				ResultSet resultSet = ps.executeQuery();
				if (resultSet.next()) {
					return new Tuple8<>(
							integerStringStringStringStringStringStringTuple7.f0,
							integerStringStringStringStringStringStringTuple7.f1,
							integerStringStringStringStringStringStringTuple7.f2,
							integerStringStringStringStringStringStringTuple7.f3,
							integerStringStringStringStringStringStringTuple7.f4,
							integerStringStringStringStringStringStringTuple7.f5,
							integerStringStringStringStringStringStringTuple7.f6,
							resultSet.getString("dept_name")
					);
				} else {
					return null;
				}
			}

			@Override
			public void close () throws Exception{
				ps.close();
				connection.close();
			}
		});

		empExProTupleStream.print();
		env.execute("tran Benchmark");

	}
}